package com.example.day04.controller.sideoutputs;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class WaterDemo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("node3", 7777);

        SingleOutputStreamOperator<WaterEntityDemo> waterEntitySingleOutputStreamOperator = source.map((String s) -> {
            String[] split = s.split(",");
            return new WaterEntityDemo(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterEntityDemo>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((element, recordTimestamp) -> element.getVc() * 1000L));

        SingleOutputStreamOperator<String> process = waterEntitySingleOutputStreamOperator.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new MyTopNpAWF());

        process.print();

        env.execute("aaa");
    }
    public static class MyTopNpAWF extends ProcessAllWindowFunction<WaterEntityDemo, String, TimeWindow> {
        @Override
        public void process(Context context, Iterable<WaterEntityDemo> elements, Collector<String> out) throws Exception {
            Map<String, Integer> map = new HashMap<>();

            for (WaterEntityDemo element : elements) {
                String id = element.getId();
                if (!map.containsKey(id)) {
                    map.put(id, 1);
                } else {
                    map.put(id, map.get(id) + 1);
                }
            }

            List<Tuple2<String, Integer>> datas = new ArrayList<>();
            for (String key : map.keySet()) {
                datas.add(Tuple2.of(key, map.get(key)));
            }

            // 降序排列
            datas.sort((o1, o2) -> o2.f1.compareTo(o1.f1)); // 使用 compareTo 而不是减法

            // 输出 Top 3 结果
            int topN = 3;
            for (int i = 0; i < Math.min(datas.size(), topN); i++) {
                Tuple2<String, Integer> data = datas.get(i);
                out.collect("Top-" + (i + 1) + ": id=" + data.f0 + ", count=" + data.f1 + "窗口的开始时间结束时间：" + context.window().getStart() + " - " + context.window().getEnd());
            }
        }
    }
}
